package server

import (
	"bufio"
	"context"
	"edata-hub/pkg/raft"
	"edata-hub/pkg/server/services"
	"encoding/csv"
	"encoding/json"
	"fmt"
	"gopkg.in/yaml.v3"
	"os"
	"strconv"
	"strings"
	"time"
)

const (
	CLUSTER_INSTALL        = "CLUSTER_INSTALL"
	CLUSTER_CREATION       = "CLUSTER_CREATION"
	PROGRESS_CREATION      = "PROGRESS_CREATION"
	PROGRESS_SEARCH        = "PROGRESS_SEARCH"
	NODE_CREATION          = "NODE_CREATION"
	ROLE_SEARCH            = "ROLE_SEARCH"
	ROLE_CREATION          = "ROLE_CREATION"
	CLUSTER_SEARCH         = "CLUSTER_SEARCH"
	NODES_SEARCH           = "NODES_SEARCH"
	NODE_STATUS_SEARCH     = "NODE_STATUS_SEARCH"
	DATACOMP_SEARCH        = "DATACOMP_SEARCH"
	DATACOMP_MODE_SEARCH   = "DATACOMP_MODE_SEARCH"
	DATACOMP_ROLE_SEARCH   = "DATACOMP_ROLE_SEARCH"
	DATACOMP_CONFIG_SEARCH = "DATACOMP_CONFIG_SEARCH"
	DATACOMP_INSTALL       = "DATACOMP_INSTALL"
	WORKER_TRANSPORT       = "WORKER_TRANSPORT"
	DATACOMP_TRANSPORT     = "DATACOMP_TRANSPORT"
	CHMOD_ALL_COMPS        = "CHMOD_ALL_COMPS"
	WORKER_START           = "WORKER_START"
	ROLE_START             = "ROLE_START"
	ROLE_STOP              = "ROLE_STOP"
	WORKER_STOP            = "WORKER_STOP"
	ENV_GENERATION         = "ENV_GENERATION"
	MODE_GENERATION        = "MODE_GENERATION"
)

func (m *Master) InitExCommand() {
	m.RServer.RegisterExCommands(CLUSTER_CREATION, "create cluster", m.CreateCluster)
	m.RServer.RegisterExCommands(PROGRESS_CREATION, "create cluster", m.CreateProgress)
	m.RServer.RegisterExCommands(NODE_CREATION, "create node", m.CreateNode)
	m.RServer.RegisterExCommands(ROLE_CREATION, "create role", m.CreateRoles)
	m.RServer.RegisterExCommands(ROLE_SEARCH, "create role", m.FindRoles)
	m.RServer.RegisterExCommands(CLUSTER_SEARCH, "find clusters", m.FindClusters)
	m.RServer.RegisterExCommands(PROGRESS_SEARCH, "find progress", m.FindProgress)
	m.RServer.RegisterExCommands(NODES_SEARCH, "find nodes", m.FindNodes)
	m.RServer.RegisterExCommands(NODE_STATUS_SEARCH, "find node status", m.FindNodeStatus)
	m.RServer.RegisterExCommands(DATACOMP_SEARCH, "find components", m.FindComponents)
	m.RServer.RegisterExCommands(DATACOMP_MODE_SEARCH, "find modes", m.FindCompModes)
	m.RServer.RegisterExCommands(DATACOMP_ROLE_SEARCH, "find roles", m.FindCompRoles)
	m.RServer.RegisterExCommands(DATACOMP_CONFIG_SEARCH, "find component config", m.FindCompConfigs)
	m.RServer.RegisterExCommands(DATACOMP_INSTALL, "execute install.sh of data component", m.InstallDataComps)
	m.RServer.RegisterExCommands(WORKER_TRANSPORT, "send the worker application to the remote server", m.WorkerTransport)
	m.RServer.RegisterExCommands(DATACOMP_TRANSPORT, "send the data component to the remote server", m.DataCompTransport)
	m.RServer.RegisterExCommands(CHMOD_ALL_COMPS, "chmod the compoennts(edh & data) to -R 777", m.ChmodAllComps)
	m.RServer.RegisterExCommands(WORKER_START, "activate the worker application", m.StartWorker)
	m.RServer.RegisterExCommands(WORKER_STOP, "deactivate the worker application", m.StopWorker)
	m.RServer.RegisterExCommands(ROLE_START, "activate the data component", m.StartRole)
	m.RServer.RegisterExCommands(ROLE_STOP, "stop the data component", m.StopRole)
	m.RServer.RegisterExCommands(ENV_GENERATION, "create the environment variable file", m.GenerateENV)
	m.RServer.RegisterExCommands(MODE_GENERATION, "create the [sa]mode file", m.GenerateMode)
}

func (m *Master) InitExCommandOptimizer() {
	m.RServer.RegisterOptimizer(m.Optimizer)
}

func (m *Master) CreateCluster(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	clusterId := args["clusterId"]
	clusterPath := fmt.Sprintf("%s/%s", m.Config.MasterConfig.ClustersPath, clusterId)
	mkDirErr := os.MkdirAll(clusterPath, 0755)
	if mkDirErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("Create Cluster %s failed:%s", clusterId, mkDirErr.Error())
		return ret
	}
	clusterInfoPath := fmt.Sprintf("%s/cluster", clusterPath)
	clusterInfo := ClusterInfo{
		ClusterId: args["clusterId"],
		Name:      args["name"],
		Comp:      args["comp"],
		Type:      args["type"],
		Mode:      args["[sa]mode"],
		Status:    0,
	}
	clusterInfoFile, createErr := os.Create(clusterInfoPath)
	if createErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not create cluster info file:%s", createErr.Error())
		return ret
	}
	defer clusterInfoFile.Close()
	infoJson, _ := json.Marshal(clusterInfo)
	_, writeErr := clusterInfoFile.Write(infoJson)
	if writeErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not write cluster info:%s", writeErr.Error())
		return ret
	}
	ret.Status = 1
	ret.Msg = fmt.Sprintf("Create Cluster %s success", clusterId)
	return ret
}

func (m *Master) CreateProgress(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	clusterId := args["clusterId"]
	progressPath := fmt.Sprintf("%s/%s", m.Config.MasterConfig.ProgressPath, clusterId)
	progressFile, createProgressFileErr := os.Create(progressPath)
	if createProgressFileErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("Create Progress of %s failed:%s", clusterId, createProgressFileErr.Error())
		return ret
	}
	_, writeErr := progressFile.Write([]byte(args["progress"]))
	if writeErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not write progress file of %s:%s", clusterId, writeErr.Error())
		return ret
	}
	ret.Status = 1
	ret.Msg = fmt.Sprintf("Create Progress of %s success", clusterId)
	return ret
}

func (m *Master) FindProgress(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	clusterId := args["clusterId"]
	progressPath := fmt.Sprintf("%s/%s", m.Config.MasterConfig.ProgressPath, clusterId)
	progressData, readErr := os.ReadFile(progressPath)
	if readErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not read progress file of %s:%s", clusterId, readErr.Error())
		return ret
	}
	progressArr := make([]string, 0)
	lines := strings.Split(string(progressData), "\n")
	for _, line := range lines {
		name_status := strings.Split(line, ",")
		step := fmt.Sprintf("{\"name\":\"%s\",\"status\":%s}", name_status[0], name_status[1])
		progressArr = append(progressArr, step)
	}
	ret.Status = 1
	ret.Msg = fmt.Sprintf("Find Progress of %s success", clusterId)
	ret.Data = "[" + strings.Join(progressArr, ",") + "]"
	return ret
}

func (m *Master) CreateNode(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	clusterId := args["clusterId"]
	n := args["node"]
	var node NodeInfo
	unmarshalNodeErr := json.Unmarshal([]byte(n), &node)
	if unmarshalNodeErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not unmarshal node info:%s", unmarshalNodeErr.Error())
		return ret
	}
	nodePath := fmt.Sprintf("%s/%s/%s", m.Config.MasterConfig.ClustersPath, clusterId, node.Ip)
	mkdirErr := os.MkdirAll(nodePath, 0755)
	if mkdirErr != nil {
		m.RServer.Msg.Info("can not create directory %s:%s", nodePath, mkdirErr.Error())
	}
	nodeInfoPath := fmt.Sprintf("%s/%s/%s/node", m.Config.MasterConfig.ClustersPath, clusterId, node.Ip)
	nodeInfoFile, _ := os.Create(nodeInfoPath)
	_, writeErr := nodeInfoFile.Write([]byte(n))
	if writeErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not write node-info %s:%s", nodeInfoPath, writeErr.Error())
		return ret
	}
	defer nodeInfoFile.Close()
	ret.Status = 1
	ret.Msg = fmt.Sprintf("create node %s success", node.Ip)
	return ret
}

func (m *Master) CreateRoles(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	clusterId := args["clusterId"]
	ip := args["ip"]
	roles := strings.Split(args["roles"], ",")
	for _, role := range roles {
		roleInfoPath := fmt.Sprintf("%s/%s/%s/%s", m.Config.MasterConfig.ClustersPath, clusterId, ip, role)
		roleInfoFile, _ := os.Create(roleInfoPath)
		defer roleInfoFile.Close()
	}
	ret.Status = 1
	ret.Msg = fmt.Sprintf("%v,create roles success", roles)
	return ret
}

func (m *Master) WorkerTransport(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	ip := args["addr"]
	port := "22"
	user_password := strings.Split(args["auths"], "/")
	user := user_password[0]
	password := user_password[1]
	shell := "uname -a"
	osTypeData, findOsTypeErr := m.ExecShell(ip, port, user, password, shell)
	if findOsTypeErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not find remote os version when transporting worker:%s", findOsTypeErr.Error())
		return ret
	}
	localAppPath := m.Config.MasterConfig.EdhCompPath
	if strings.Contains(string(osTypeData), "Linux") {
		localAppPath = localAppPath + "/worker"
	} else {
		localAppPath = localAppPath + "/worker.exe"
	}
	uploadWorkerErr := m.UploadFile(ip, port, user, password, localAppPath, m.Config.MasterConfig.RemotePath)
	if uploadWorkerErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not upload app when transporting worker:%s", uploadWorkerErr.Error())
		return ret
	}
	// 将配置文件application。yml写入目标节点
	m.Config.WorkerConfig.Ip = ip
	data, _ := yaml.Marshal(&m.Config)
	remoteConfPath := m.Config.MasterConfig.RemotePath + "/configs/application.yml"
	uploadWorkerConfigErr := m.WriteBytesToRemoteFile(ip, port, user, password, data, remoteConfPath)
	if uploadWorkerConfigErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not upload worker config when transporting worker:%s", uploadWorkerConfigErr.Error())
		return ret
	}
	seviceScript := []string{
		"[Unit]",
		"Description=Start the worker via the systemctl",
		"[Service]",
		"Type=simple",
		"Restart=on-failure",
		"User=root",
		fmt.Sprintf("ExecStart=/home/smart/edh/worker -ip=%s -confPath=%s/configs/application.yml", ip, m.Config.MasterConfig.RemotePath),
		"[Install]",
		"WantedBy=multi-user.target\n",
	}
	servicePath := "/lib/systemd/system/worker.service"
	writeServiceErr := m.WriteBytesToRemoteFile(ip, port, user, password, []byte(strings.Join(seviceScript, "\n")), servicePath)
	if writeServiceErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not write remote worker service:%s", writeServiceErr.Error())
		return ret
	}
	ret.Msg = fmt.Sprintf("transport worker to %s success", ip)
	ret.Status = 1
	return ret
}

func (m *Master) DataCompTransport(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	ip := args["addr"]
	port := "22"
	user_password := strings.Split(args["auths"], "/")
	user := user_password[0]
	password := user_password[1]
	compDir := "./components/" + args["comp"] + "/"
	compIndex := compDir + "edh_index"
	file, openIndexErr := os.Open(compIndex)
	defer file.Close()
	if openIndexErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not open comps index when transporting data components:%s", openIndexErr.Error())
		return ret
	}
	scanner := bufio.NewScanner(file)
	for scanner.Scan() {
		filename := scanner.Text()
		localPath := compDir + filename
		remotePath := m.Config.MasterConfig.RemotePath + "/comps/" + args["comp"]
		uploadCompsErr := m.UploadFile(ip, port, user, password, localPath, remotePath)
		if uploadCompsErr != nil {
			ret.Status = -1
			ret.Msg = fmt.Sprintf("can not upload comps:%s", uploadCompsErr.Error())
			return ret
		}
	}
	ret.Msg = fmt.Sprintf("transport data component to %s success", ip)
	ret.Status = 1
	return ret
}

func (m *Master) ChmodAllComps(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	ip := args["addr"]
	port := "22"
	user_password := strings.Split(args["auths"], "/")
	user := user_password[0]
	password := user_password[1]
	chmodEDHPathShell := fmt.Sprintf("chmod -R 777 %s", m.Config.MasterConfig.RemotePath)
	_, chmodEDHPathErr := m.ExecShell(ip, port, user, password, chmodEDHPathShell)
	if chmodEDHPathErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not chmod remote EDH path:%s", chmodEDHPathErr.Error())
		return ret
	}
	chmodWorkerServiceShell := fmt.Sprintf("chmod -R 777 %s", "/lib/systemd/system/worker.service")
	_, chmodWorkerServiceErr := m.ExecShell(ip, port, user, password, chmodWorkerServiceShell)
	if chmodWorkerServiceErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not chmod worker service:%s", chmodWorkerServiceErr.Error())
		return ret
	}
	ret.Status = 1
	ret.Msg = fmt.Sprintf("chmod remote components of %s success", ip)
	return ret
}

func (m *Master) InstallDataComps(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	ip := args["addr"]
	port := m.Config.WorkerConfig.RpcPort
	comp := args["comp"]
	shell := fmt.Sprintf("%s/comps/%s/%s", m.Config.MasterConfig.RemotePath, comp, "install.sh")
	params := make(map[string]string)
	params["shell"] = shell
	req := &services.CommandWorkerReq{
		Command: "EXEC_SHELL",
		Params:  params,
	}
	client, createErr := m.CreateWorkerRPCClient(ip, port)
	if createErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not create worker rpc client:%s", createErr.Error())
		return ret
	}
	stream, sendCmdErr := client.OnHandleCommand(context.Background(), req)
	if sendCmdErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not send command %s to worker %s:%s", shell, ip, sendCmdErr.Error())
		return ret
	}
	resp, recvErr := stream.Recv()
	if recvErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not receive response from worker %s:%s", ip, recvErr.Error())
		return ret
	}
	ret.Status = resp.Status
	ret.Msg = resp.Msg
	return ret
}

func (m *Master) StartRole(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	ip := args["addr"]
	port := m.Config.WorkerConfig.RpcPort
	comp := args["comp"]
	role := args["role"]
	shell := fmt.Sprintf("%s/comps/%s/%s %s", m.Config.MasterConfig.RemotePath, comp, "start.sh", role)
	//shell = fmt.Sprintf("%s/comps/%s/%s %s", m.Config.MasterConfig.RemotePath, comp, "test.sh", role)
	params := make(map[string]string)
	params["shell"] = shell
	req := &services.CommandWorkerReq{
		Command: "EXEC_SHELL",
		Params:  params,
	}
	client, createErr := m.CreateWorkerRPCClient(ip, port)
	if createErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not create worker rpc client:%s", createErr.Error())
		return ret
	}
	stream, sendCmdErr := client.OnHandleCommand(context.Background(), req)
	if sendCmdErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not send command %s to worker %s:%s", shell, ip, sendCmdErr.Error())
		return ret
	}
	resp, recvErr := stream.Recv()
	if recvErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not receive response from worker %s:%s", ip, recvErr.Error())
		return ret
	}
	ret.Status = resp.Status
	ret.Msg = resp.Msg
	return ret
}

func (m *Master) StopRole(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	ip := args["addr"]
	port := m.Config.WorkerConfig.RpcPort
	comp := args["comp"]
	role := args["role"]
	shell := fmt.Sprintf("%s/comps/%s/%s %s", m.Config.MasterConfig.RemotePath, comp, "stop.sh", role)
	params := make(map[string]string)
	params["shell"] = shell
	req := &services.CommandWorkerReq{
		Command: "EXEC_SHELL",
		Params:  params,
	}
	client, createErr := m.CreateWorkerRPCClient(ip, port)
	if createErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not create worker rpc client:%s", createErr.Error())
		return ret
	}
	stream, sendCmdErr := client.OnHandleCommand(context.Background(), req)
	if sendCmdErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not send command %s to worker %s:%s", shell, ip, sendCmdErr.Error())
		return ret
	}
	resp, recvErr := stream.Recv()
	if recvErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not receive response from worker %s:%s", ip, recvErr.Error())
		return ret
	}
	ret.Status = resp.Status
	ret.Msg = resp.Msg
	return ret
}

func (m *Master) StartWorker(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	ip := args["addr"]
	port := "22"
	user_password := strings.Split(args["auths"], "/")
	user := user_password[0]
	password := user_password[1]
	checkStatusShell := "systemctl status worker.service"
	statusInfo, _ := m.ExecShell(ip, port, user, password, checkStatusShell)
	if strings.Contains(string(statusInfo), "active (running)") {
		ret.Status = 1
		ret.Msg = fmt.Sprintf("worker %s is running:\n%s", ip, string(statusInfo))
		return ret
	} else {
		startAndCheckShell := "systemctl daemon-reload && systemctl start worker.service && systemctl status worker.service"
		startInfo, _ := m.ExecShell(ip, port, user, password, startAndCheckShell)
		if strings.Contains(string(startInfo), "active (running)") {
			//在此处验证是否能与Worker通信，
			checkSum := 10
			checkNum := 1
			for checkNum = 1; checkNum <= checkSum; checkNum++ {
				client, createErr := m.CreateWorkerRPCClient(ip, port)
				if createErr != nil {
					m.RServer.Msg.Info("[%d/%d] can not create worker rpc client:%s", checkNum, checkSum, createErr.Error())
					time.Sleep(5 * time.Second)
					continue
				}
				checkRPCConnScript := fmt.Sprintf("echo 'Hello Worker'")
				params := make(map[string]string)
				params["shell"] = checkRPCConnScript
				req := &services.CommandWorkerReq{
					Command: "EXEC_SHELL",
					Params:  params,
				}
				stream, sendCmdErr := client.OnHandleCommand(context.Background(), req)
				if sendCmdErr != nil {
					m.RServer.Msg.Info("[%d/%d] can not send command %s to worker %s:%s", checkNum, checkSum, checkRPCConnScript, ip, sendCmdErr.Error())
					time.Sleep(5 * time.Second)
					continue
				}
				resp, recvErr := stream.Recv()
				if recvErr != nil {
					m.RServer.Msg.Info("[%d/%d] can not receive response from worker %s:%s", checkNum, checkSum, ip, recvErr.Error())
					continue
				}
				if resp.Status == 1 {
					m.RServer.Msg.Info("[%d/%d] rpc connection with worker %s established", checkNum, checkSum, ip)
					break
				}
			}
			if checkNum > checkSum {
				ret.Status = -1
				ret.Msg = fmt.Sprintf("[%d/%d] can not establish rpc connection with worker %s", checkNum, checkSum, ip)
				return ret
			}
			ret.Status = 1
			ret.Msg = fmt.Sprintf("start worker %s success:\n%s", ip, string(startInfo))
			return ret
		} else {
			ret.Status = -1
			ret.Msg = fmt.Sprintf("can not start worker %s:\n%s", ip, string(startInfo))
		}
	}
	return ret
}

func (m *Master) StopWorker(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	ip := args["addr"]
	port := "22"
	user_password := strings.Split(args["auths"], "/")
	user := user_password[0]
	password := user_password[1]
	stopWorkerShell := "systemctl stop worker.service && systemctl status worker.service"
	stopInfo, _ := m.ExecShell(ip, port, user, password, stopWorkerShell)
	if strings.Contains(string(stopInfo), "inactive (dead)") {
		ret.Status = 1
		ret.Msg = fmt.Sprintf("stopping worker %s success \n %s", ip, string(stopInfo))
	}
	return ret
}

func (m *Master) GenerateMode(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	ip := args["addr"]
	port := "22"
	user_password := strings.Split(args["auths"], "/")
	user := user_password[0]
	password := user_password[1]
	remote_mode_file := m.Config.MasterConfig.RemotePath + "/comps/" + args["comp"] + "/[sa]mode"
	data := []byte(args["[sa]mode"])
	err := m.WriteBytesToRemoteFile(ip, port, user, password, data, remote_mode_file)
	if err != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not generate [sa]mode to remote path:%s", err.Error())
		return ret
	}
	ret.Msg = fmt.Sprintf("generate [sa]mode to remote path %s success", remote_mode_file)
	ret.Status = 1
	return ret
}

func (m *Master) GenerateENV(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	ip := args["addr"]
	port := "22"
	user_password := strings.Split(args["auths"], "/")
	user := user_password[0]
	password := user_password[1]
	remote_env_file := m.Config.MasterConfig.RemotePath + "/comps/" + args["comp"] + "/[sa]config.env"
	data := []byte(args["env"])
	err := m.WriteBytesToRemoteFile(ip, port, user, password, data, remote_env_file)
	if err != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not generate env to remote path:%s", err.Error())
		return ret
	}
	ret.Msg = fmt.Sprintf("generate env to remote path %s success", remote_env_file)
	ret.Status = 1
	return ret
}

func (m *Master) FindComponents(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	file, open_comps_index_err := os.Open("components/index")
	defer file.Close()
	if open_comps_index_err != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not find components index when finding components:%s", open_comps_index_err.Error())
		return ret
	}
	reader := csv.NewReader(file)
	records, read_comp_index_err := reader.ReadAll()
	if read_comp_index_err != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not read components index when finding components:%s", read_comp_index_err.Error())
		return ret
	}
	var comps []Comp
	for _, record := range records {
		stat, _ := strconv.Atoi(record[2])
		comp := Comp{
			Name:   record[0],
			Type:   record[1],
			Status: stat,
		}
		comps = append(comps, comp)
	}
	ret.Status = 1
	ret.Msg = "find components success"
	comps_json, _ := json.Marshal(comps)
	ret.Data = string(comps_json)
	return ret
}

func (m *Master) FindCompModes(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	filePath := "components/" + args["comp"] + "/[sa]mode"
	file, open_comp_mode_err := os.Open(filePath)
	if open_comp_mode_err != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not open component modes when finding modes:%s", open_comp_mode_err.Error())
		return ret
	}
	defer file.Close()
	reader := csv.NewReader(file)
	records, read_comp_mode_err := reader.ReadAll()
	if read_comp_mode_err != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not read component modes when finding modes:%s", read_comp_mode_err.Error())
	}
	var modes []string
	for _, record := range records {
		modes = append(modes, record[0])
	}
	ret.Status = 1
	ret.Msg = "find component modes success"
	ret.Data = strings.Join(modes, ",")
	return ret
}

func (m *Master) FindCompRoles(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	filePath := "components/" + args["comp"] + "/edh_roles_" + args["[sa]mode"]
	file, open_comp_role_err := os.Open(filePath)
	if open_comp_role_err != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not open component roles when finding roles:%s", open_comp_role_err.Error())
		return ret
	}
	defer file.Close()
	reader := csv.NewReader(file)
	records, read_comp_role_err := reader.ReadAll()
	if read_comp_role_err != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not read component roles when finding roles:%s", read_comp_role_err.Error())
		return ret
	}
	var roles []string
	for _, record := range records {
		roles = append(roles, record[0])
	}
	ret.Status = 1
	ret.Msg = "find component roles success"
	ret.Data = strings.Join(roles, ",")
	return ret
}

func (m *Master) FindCompConfigs(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	filePath := "components/" + args["comp"] + "/edh_config_" + args["[sa]mode"] + ".env"
	file, open_comp_config_err := os.Open(filePath)
	if open_comp_config_err != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not open component configs when finding config:%s", open_comp_config_err.Error())
		return ret
	}
	defer file.Close()
	scanner := bufio.NewScanner(file)
	var configs []string
	for scanner.Scan() {
		text := strings.TrimSpace(scanner.Text())
		if len(text) != 0 {
			configs = append(configs, scanner.Text())
		}
	}
	ret.Status = 1
	ret.Msg = "find component config success"
	ret.Data = strings.Join(configs, "\n")
	return ret
}

func (m *Master) FindClusters(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	clusterPath := m.Config.MasterConfig.ClustersPath
	datas := make([]string, 0)
	clusterDir, openErr := os.Open(clusterPath)
	if openErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not open cluster dir %s:%s", clusterPath, openErr.Error())
		return ret
	}
	defer clusterDir.Close()
	dirs, dirErr := clusterDir.ReadDir(-1) //-1表示所有
	if dirErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not read dir %s:%s", clusterPath, dirErr.Error())
		return ret
	}
	for _, dir := range dirs {
		clusterInfoFile := fmt.Sprintf("%s/%s/%s", clusterPath, dir.Name(), "cluster")
		infoData, readErr := os.ReadFile(clusterInfoFile)
		if readErr != nil {
			ret.Status = -1
			ret.Msg = fmt.Sprintf("can not read cluster info file %s:%s", clusterInfoFile, readErr.Error())
			return ret
		}
		datas = append(datas, string(infoData))
	}
	ret.Status = 1
	ret.Msg = "find clusters success"
	ret.Data = "[" + strings.Join(datas, ",") + "]"
	return ret
}

func (m *Master) FindNodes(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	cluster := fmt.Sprintf("%s/%s", m.Config.MasterConfig.ClustersPath, args["clusterId"])
	datas := make([]string, 0)
	clusterDir, openErr := os.Open(cluster)
	if openErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not open cluster dir %s:%s", cluster, openErr.Error())
		return ret
	}
	defer clusterDir.Close()
	nodeDirs, dirsErr := clusterDir.ReadDir(-1)
	if dirsErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not read dir %s:%s", cluster, dirsErr.Error())
		return ret
	}
	// 遍历每个节点
	for _, nodeDir := range nodeDirs {
		if !nodeDir.IsDir() {
			continue
		}
		nodeInfoPath := fmt.Sprintf("%s/%s/%s", cluster, nodeDir.Name(), "node")
		nodeInfoData, readErr := os.ReadFile(nodeInfoPath)
		if readErr != nil {
			ret.Status = -1
			ret.Msg = fmt.Sprintf("can not read node info file %s", nodeInfoPath, readErr.Error())
			return ret
		}
		datas = append(datas, string(nodeInfoData))
	}
	ret.Status = 1
	ret.Data = "[" + strings.Join(datas, ",") + "]"
	ret.Msg = "find cluster nodes success"
	return ret
}

func (m *Master) FindNodeStatus(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	ip := args["ip"]
	datas := make([]string, 0)
	workerInfo := m.RServer.Workers[ip]
	passed := int(time.Now().Sub(workerInfo.TickTok).Seconds())
	passed_json := fmt.Sprintf("\"timeout\":\"%d秒之前\"", passed)
	status_json := fmt.Sprintf("\"status\":%d", 1)
	if passed > m.Config.WorkerConfig.Timeout {
		passed_json = fmt.Sprintf("\"timeout\":\"%d秒之前（超时）\"", passed)
		status_json = fmt.Sprintf("\"status\":%d", 0)
	}
	datas = append(datas, status_json)
	datas = append(datas, passed_json)
	datas = append(datas, fmt.Sprintf("\"usedMem\":%d", workerInfo.UsedMem/1024/1024))
	datas = append(datas, fmt.Sprintf("\"totalMem\":%d", workerInfo.TotalMem/1024/1024))
	datas = append(datas, fmt.Sprintf("\"usedDisk\":%d", workerInfo.UsedDisk/1024/1024))
	datas = append(datas, fmt.Sprintf("\"totalDisk\":%d", workerInfo.TotalDisk/1024/1024))
	ret.Status = 1
	ret.Data = fmt.Sprintf("{" + strings.Join(datas, ",") + "}")
	ret.Msg = "find node status success"
	return ret
}

func (m *Master) FindRoles(args map[string]string) raft.CommandExecRet {
	ret := raft.CommandExecRet{
		Status: 0,
		Msg:    "",
	}
	clusterId := args["clusterId"]
	ip := args["ip"]
	port := m.Config.WorkerConfig.RpcPort
	comp := args["comp"]
	nodeDirPath := fmt.Sprintf("%s/%s/%s", m.Config.MasterConfig.ClustersPath, clusterId, ip)
	nodeDir, openErr := os.Open(nodeDirPath)
	if openErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not open node dir %s:%s", nodeDirPath, openErr.Error())
		return ret
	}
	defer nodeDir.Close()
	roleDirs, readErr := nodeDir.ReadDir(-1)
	if readErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not read dir %s:%s", nodeDir, readErr.Error())
		return ret
	}

	shell := m.GenCheckRoleStatusScript(comp)
	params := make(map[string]string)
	params["shell"] = shell
	req := &services.CommandWorkerReq{
		Command: "EXEC_SHELL",
		Params:  params,
	}
	client, createClientErr := m.CreateWorkerRPCClient(ip, port)
	if createClientErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not create worker rpc client:%s", createClientErr.Error())
		return ret
	}
	stream, sendCmdErr := client.OnHandleCommand(context.Background(), req)
	if sendCmdErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not send command:%s", sendCmdErr.Error())
		return ret
	}
	resp, recErr := stream.Recv()
	if recErr != nil {
		ret.Status = -1
		ret.Msg = fmt.Sprintf("can not recv command:%s", recErr.Error())
		return ret
	}
	datas := make([]string, 0)
	for _, roleDir := range roleDirs {
		if roleDir.IsDir() || roleDir.Name() == "node" {
			continue
		}
		if strings.Contains(strings.ToLower(resp.Msg), roleDir.Name()) {
			datas = append(datas, fmt.Sprintf("{\"name\":\"%s\",\"status\":%d}", roleDir.Name(), 1))
		} else {
			datas = append(datas, fmt.Sprintf("{\"name\":\"%s\",\"status\":%d}", roleDir.Name(), 0))
		}
	}
	ret.Status = 1
	ret.Data = "[" + strings.Join(datas, ",") + "]"
	ret.Msg = resp.Msg
	return ret
}
